package com.atguigu.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.bean.TableProcess;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.JdbcUtil;
import com.atguigu.realtime.util.RedisUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2022/2/11 11:03
 * 自定义Phoenix sink
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    private Connection conn;
    private ValueState<Boolean> tableCreatedState;
    private ValueState<String> upsetSqlState;
    private Jedis redisClient;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 建立到Phoenix的连接
        String driver = Constant.PHOENIX_DRIVER;
        String url = Constant.PHOENIX_URL;
        conn = JdbcUtil.getConnection(driver, url);
        
        tableCreatedState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("tableCreatedState", Boolean.class));
        
        upsetSqlState = getRuntimeContext().getState(new ValueStateDescriptor<String>("upsetSqlState", String.class));
        
        redisClient = RedisUtil.getRedisClient();
        
    }
    
    @Override
    public void close() throws Exception {
        // 关闭连接
        if (conn != null && !conn.isClosed()) {
            conn.close();
        }
        
        if (redisClient != null) {
            redisClient.close();
        }
    }
    
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value,
                       Context context) throws Exception {
        // 1. 建表
        checkTable(value);
        // 2. 把这条数据插入到表中
        write2Phoenix(value);
        
        // 3. 更新redis中缓存维度(增加幻化之后新增的代码)
        updateCache(value);
        
    }
    
    private void updateCache(Tuple2<JSONObject, TableProcess> value) {
        /*
        两种更新方案:
        1. 粗暴
            如果发现维度有变化, 则直接去redis删除缓存
                dwm层读取缓存没有了, 自动去数据库读取, 一定是最新的
        
        2. 优雅的
            把缓存中的数据更新到最新的维度信息
         */
        JSONObject data = value.f0;  // 从mysql直接出来, 里面的列名(字段)都是小写
        TableProcess tp = value.f1;
        
        String key = tp.getSink_table() + ":" + data.getLong("id");
        
        if ("update".equals(tp.getOperate_type()) && redisClient.exists(key)) {
            //如果这次是更新, 更新这个key在redis存在, 才需要更新缓存
            // 应该把data中的key应该关不缓存大写
            JSONObject upperData = new JSONObject();
            for (Map.Entry<String, Object> entry : data.entrySet()) {
                String k = entry.getKey().toUpperCase();
                Object v = entry.getValue();
    
                upperData.put(k, v);
            }
    
            redisClient.setex(key, Constant.DIM_CACHE_TIMEOUT, upperData.toJSONString());  // key更新, 则ttl失效, 相当于ttl配置,所以需要对ttl重新做设置
        }
        
        // 粗暴
       // redisClient.del(key);
        
    }
    
    private void write2Phoenix(Tuple2<JSONObject, TableProcess> value) throws SQLException, IOException {
        JSONObject data = value.f0;
        TableProcess tp = value.f1;
        
        // 如果主键不存在, 就是insert, 如果主键已经存在, 则update
        //  upset into user(id, name, age)values(?,?,?)
        
        String sql = upsetSqlState.value();
        
        if (sql == null) {
            sql = "upsert into " +
                tp.getSink_table() +
                "(" + tp.getSink_columns() +  // id,coupon_id,range_type,range_id
                ")values(" +
                tp.getSink_columns().replaceAll("[^,]+", "?") +
                ")";
            upsetSqlState.update(sql);
        }
        
        //        System.out.println(tp.getSink_table() + " 插入语句 " + sql);
        PreparedStatement ps = conn.prepareStatement(sql);
        // 使用具体的值对占位符进行替换:
        
        String[] columns = tp.getSink_columns().split(",");
        for (int i = 0; i < columns.length; i++) {
            String columnName = columns[i];
            // jsonObject对象获取数据, 如果类型不是字符串, 直接获取字符串会出问题
            Object v = data.get(columnName); //v有可能是null
            // 向Phoenix写入的时候, 必须是字符串
            ps.setString(i + 1, v == null ? "" : v.toString());   // v 是null  -> "null"
            
        }
        ps.execute();
        conn.commit();
        ps.close();
    }
    
    private void checkTable(Tuple2<JSONObject, TableProcess> t) throws SQLException, IOException {
        
        if (tableCreatedState.value() == null) {
            TableProcess tp = t.f1;
        /*
        根据表名和列通过Phoenix在hbase中建表
         */
            // 1. 定义sql语句
            // create table if no exists a(id varchar, name varchar, age varchar, constraint pk primary key(id, name)) SALT_BUCKETS = 4
            // id,coupon_id,range_type,range_id
            String sql = "create table if not exists " +
                tp.getSink_table() +
                "(" +
                tp.getSink_columns().replaceAll(",", " varchar,") +
                " varchar, constraint pk primary key(" +
                (tp.getSink_pk() == null ? "id" : tp.getSink_pk()) +  // 默认使用id作为主键
                "))" + (tp.getSink_extend() == null ? "" : tp.getSink_extend());
            // 2. 根据连接对象, 获取一个 PrepareState
            System.out.println("建表语句: " + sql);
            PreparedStatement ps = conn.prepareStatement(sql);
            // sql可能会有占位符, 还需要对占位符进行赋值   ????
            // 3. 执行sql语句
            ps.execute();
            conn.commit();
            ps.close();
            
            tableCreatedState.update(true);
        }
    }
}
